package com.zookeeper.locks;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/**
 * https://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper/
 * @author Tony
 *
 */
public class Locks implements Watcher {

	protected static ZooKeeper zk = null;
	protected static Integer mutex;
	int sessionTimeout = 10000;
	protected String root;

	public Locks(String connectString, String root) {
		if (zk == null) {
			try {
				System.out.println("创建一个新的连接:");
				zk = new ZooKeeper(connectString, sessionTimeout, this);
				mutex = new Integer(-1);
			} catch (IOException e) {
				zk = null;
			}
		}
		this.root = root;
		try {
			Stat s = zk.exists(root, false);
			if (s == null) {
				zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,	CreateMode.PERSISTENT);
			}
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	String myZnode;

	void getLock() throws KeeperException, InterruptedException {
		//客户端调用getChildren获取所有已经创建的子节点列表，不用注册任务Wather
		List<String> list = zk.getChildren(root, false);
		String[] nodes = list.toArray(new String[list.size()]);
		Arrays.sort(nodes);
		//判断当前新增的节点是否
		if (myZnode.equals(root + "/" + nodes[0])) {
			doAction();
		} else {
			waitForLock(nodes[0]);
		}
	}

	void waitForLock(String lower) throws InterruptedException, KeeperException {
		Stat stat = zk.exists(root + "/" + lower, true);
		if (stat != null) {
			mutex.wait();
		} else {
			getLock();
		}
	}

	
	void check() throws InterruptedException, KeeperException {
		//1、客户端调用create创建临时顺序节点
		myZnode = zk.create(root + "/lock_", new byte[0],ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
		getLock();
	}

	/**
	 * 执行其他任务
	 */
	private void doAction() {
		System.out.println("同步队列已经得到同步，可以开始执行后面的任务了");
	}

	@Override
	public void process(WatchedEvent event) {
		synchronized (mutex) {
			mutex.notify();
		}
	}
	
	
	
	 public static void main(String[] args) {
	        String connectString = "127.0.0.1:2181";
	        Locks lk = new Locks(connectString, "/shared_locks");
	        try {
	            lk.check();
	        } catch (InterruptedException e) {
	        	e.printStackTrace();
	        } catch (KeeperException e) {
	          e.printStackTrace();
	        }
	    }

}
